package cn.edu.xmu.datamining.tangzk.mralgos.parallelbfs;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import cn.edu.xmu.datamining.tangzk.util.Joiner;

public class ParalleBFSReducer extends
		Reducer<IntWritable, GraphNodeWritable, IntWritable, Text> {
	protected void reduce(
			IntWritable key,
			java.lang.Iterable<GraphNodeWritable> values,
			org.apache.hadoop.mapreduce.Reducer<IntWritable, GraphNodeWritable, IntWritable, Text>.Context context)
			throws java.io.IOException, InterruptedException {
		GraphNodeWritable graphNode = null;
		int distance = Integer.MAX_VALUE;
		for (GraphNodeWritable val : values) {
			if (val.isNode()) {
				graphNode = new GraphNodeWritable(val);
			} else {
				distance = Math.min(distance, val.getDistance());
			}
		}
		if (distance < graphNode.getDistance()) {
			graphNode.setDistance(distance);
			context.getCounter("UserDefined", ParalleBFSDriver.ITER_COUNTER)
					.increment(1);
		}

		StringBuilder sb = new StringBuilder();
		if (graphNode.getDistance() == Integer.MAX_VALUE) {
			sb.append("MAX");
		} else {
			sb.append(String.valueOf(graphNode.getDistance()));
		}
		sb.append(":");
		sb.append(Joiner.on(graphNode.getAdjNodes().iterator(), ","));
		context.write(key, new Text(sb.toString()));
	};
}
